package com.fanxl.flink.datastream.watermark;

import com.fanxl.flink.datastream.transformation.Access;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;


import java.time.LocalDateTime;

/**
 * @author Dell
 * @Title:
 * @Description: 请写注释类
 * @date 2022/6/28
 */
public class CustomerStatusChangedWatermark extends AscendingTimestampExtractor<Access> {
    @Override
    public long extractAscendingTimestamp(Access customerStatusChangedEvent) {
        return customerStatusChangedEvent.getTime();
    }


    public static void main(String[] args) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        DataStream<Access> stream = env.fromElements(
                new Access(1012525200000l, "1", (double) 2),
                new Access(1012525200006l, "4", (double) 2),
                new Access(1012525200003l, "2", (double) 2),
                new Access(1012525200005l, "3", (double) 2)
        );

        DataStream<Long> watermarkStream = stream
                .assignTimestampsAndWatermarks(new CustomerStatusChangedWatermark())
                .map(new MapFunction<Access, Long>() {
                    @Override
                    public Long map(Access p) throws Exception {
                        return p.getTime();
                    }
                });


        watermarkStream.print();
        try {
            env.execute("按事件时间排序");
        } catch (Exception ex) {

        }
    }
}
